package com.example.demo.gateway.predicate;

import static com.example.demo.gateway.core.consts.GatewayConst.CACHE_REQUEST_BODY_OBJECT_KEY;

import java.time.Duration;
import java.util.List;
import java.util.function.Function;
import java.util.function.Predicate;

import org.springframework.cloud.gateway.handler.AsyncPredicate;
import org.springframework.cloud.gateway.handler.predicate.AbstractRoutePredicateFactory;
import org.springframework.cloud.gateway.support.ServerWebExchangeUtils;
import org.springframework.http.HttpHeaders;
import org.springframework.http.codec.HttpMessageReader;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.function.server.HandlerStrategies;
import org.springframework.web.reactive.function.server.ServerRequest;
import org.springframework.web.server.ServerWebExchange;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import reactor.core.Disposable;
import reactor.core.publisher.Mono;


/**
 * @Auther: miaoguoxin
 * @Date: 2019/3/25 0025 10:47
 * @Description: 缓存请求体数据predicate
 */
@Slf4j
@Component
public class ReadBodyRoutePredicateFactory extends AbstractRoutePredicateFactory<ReadBodyRoutePredicateFactory.Config> {
    private static final List<HttpMessageReader<?>> messageReaders = HandlerStrategies
            .withDefaults().messageReaders();

    public ReadBodyRoutePredicateFactory() {
        super(Config.class);
    }


    @Override
    public Predicate<ServerWebExchange> apply(Config config) {
        throw new UnsupportedOperationException(
                "ReadBodyRoutePredicateFactory is only async.");
    }

    public AsyncPredicate<ServerWebExchange> applyAsync(Config config) {
        return exchange -> {
            ServerHttpRequest request = exchange.getRequest();
            HttpHeaders headers = request.getHeaders();
            log.info("请求headers:{}",headers);
            String rawPath = request.getURI().getRawPath();
            //兼容没有content-type的情况
            String contentTypeStr = headers.getFirst("Content-Type");
            if (contentTypeStr == null) {
                log.warn("consider your Content-Type empty?---" + rawPath);
                return Mono.just(Boolean.TRUE);
            }
            //文件上传直接跳过
            if (contentTypeStr.startsWith("multipart/form-data")) {
                return Mono.just(Boolean.TRUE);
            }
           // String cachedBody = exchange.getAttribute(CACHE_REQUEST_BODY_OBJECT_KEY);
//            BodyInserter bodyInserter = BodyInserters.fromPublisher(modifiedBody, String.class);
//            //因为请求体可能是null，这里自定义个缓存输出
//            CustomCacheBodyOutputMessage outputMessage = new CustomCacheBodyOutputMessage(exchange, headers);
            return ServerWebExchangeUtils.cacheRequestBodyAndRequest(exchange,
                    (serverHttpRequest) -> ServerRequest
                            .create(exchange.mutate().request(serverHttpRequest)
                                    .build(), messageReaders)
                            .bodyToMono(String.class)
                            .doOnNext(objectValue -> exchange.getAttributes().put(
                                    CACHE_REQUEST_BODY_OBJECT_KEY, objectValue))
                            .map(objectValue -> Boolean.TRUE));
        };
    }

    @Data
    public static class Config {
        //需要跳过的uri
        private String shouldSkipUri;

    }

    public static void main(String[] args) {
        Disposable subscribe = Mono.delay(Duration.ofSeconds(4))
                .flatMap(new Function<Long, Mono<String>>() {
                    @Override
                    public Mono<String> apply(Long aLong) {
                        System.out.println("fdgdfgdfg");

                        return Mono.just(aLong+"");
                    }
                })
                .subscribe(s -> System.out.println(s));

        while(!subscribe.isDisposed()) {
        }
        long start = System.currentTimeMillis();
        Disposable disposable = Mono.delay(Duration.ofSeconds(2)).subscribe(n -> {
            System.out.println("生产数据源："+ n);
            System.out.println("当前线程ID："+ Thread.currentThread().getId() + ",生产到消费耗时："+ (System.currentTimeMillis() - start));
        });
        while(!disposable.isDisposed()) {
        }
    }
}
